package cn.ac.iscas.service.transformer;

import cn.ac.iscas.base.utils.ParameterUtils;
import cn.ac.iscas.base.utils.SpringUtil;
import cn.ac.iscas.domain.flink.FlinkColumnType;
import cn.ac.iscas.domain.source.FlinkSourceColumn;
import cn.ac.iscas.service.queue.flink.FlinkTableQueueService;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author LJian
 * @version 1.0
 * @description: TODO
 * @date 2022/12/7 13:32
 */
public class ToggleCaseTransformer {
    private final static String ID = "togglecase_trans";
    private final static String COLUMN_KEY = "column";
    private final static String LABEL_KEY = "label";
    private final static String OPERATOR_KEY = "operator";
    private String jobId;
    private String componentId;
    private String frontTableName;
    private List<FlinkSourceColumn> frontColumns;

    private String currentTableName;
    private List<FlinkSourceColumn> currentColumns;

    public ToggleCaseTransformer(String jobId, String taskId) {
        this.jobId = jobId;
        this.componentId = taskId;
    }

    public ToggleCaseTransformer(String jobId, String taskId, String frontTableName, List<FlinkSourceColumn> frontColumns) {
        this.jobId = jobId;
        this.componentId = taskId;
        this.frontTableName = frontTableName;
        this.frontColumns = frontColumns;
        this.currentColumns = frontColumns;
        this.currentTableName = String.format("%s.%s.%s_%s", jobId, jobId, ID, UUID.randomUUID().toString().substring(0,6));
    }


    public ToggleCaseTransformer() {

    }

    //SELECT UPPER(interest) as upper_val FROM c.c.ods_classtable;
    public String getSql(Map<String, Object> parameters) {
        //String tmpTableName = this.currentTableName + "tmp";
        String createViewSql = String.format("CREATE VIEW %s AS ", this.currentTableName);
        String column = ParameterUtils.getRequiredString(parameters, COLUMN_KEY);
        String operator = ParameterUtils.getRequiredString(parameters, OPERATOR_KEY);
        String label = ParameterUtils.getRequiredString(parameters, LABEL_KEY);
        String toggleSql = "";

        if (column.isEmpty() || operator.isEmpty() || label.isEmpty()) {
            return null;
        }

        switch (operator.toUpperCase(Locale.ROOT)) {
            case "UPPER":
            case "LOWER":
                toggleSql = String.format(" %s(%s) AS %s ", operator, column, label);
                break;
            default:
                break;
        }


        String selectSql = String.format(" ( SELECT *, %s FROM %s ) ", toggleSql, frontTableName);
        createViewSql += selectSql;

        FlinkSourceColumn newColumn = new FlinkSourceColumn();
        newColumn.setType(FlinkColumnType.STRING.getName());
        newColumn.setName(label);
        newColumn.setPrimaryKey(false);
        this.currentColumns.add(newColumn);

        updateSource();
        return createViewSql;
    }


    public void updateSource() {
        FlinkTableQueueService flinkTableQueueService = SpringUtil.getBean(FlinkTableQueueService.class);
        Map<String, List<FlinkSourceColumn>> tableColumns = new ConcurrentHashMap<>(8);
        tableColumns.put(this.currentTableName, this.currentColumns);
        flinkTableQueueService.put(this.jobId, this.componentId, tableColumns);
    }
}
